package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.bean.ProvinceStats;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/4/24 9:10
 */
public class ProvinceStatsApp_TVF extends BaseSQLApp {
    public static void main(String[] args) {
        new ProvinceStatsApp_TVF().init("ProvinceStatsApp", 4003, 1, "ProvinceStatsApp");
    }
    
    @Override
    public void handle(StreamTableEnvironment tEnv) {
        
        // 1. 从dwm_order_wide消费数据:通过ddel建立动态表与这个topic进行关联
        /*
        sql中的方法调用在as前还是在as后?
            在ddl中定义字段的时候, 是在as后
            
            在sqlQuery中在as 前(值是计算得到, 算完起个别名)
         */
        tEnv.executeSql("create table ow(" +
                            "   province_id bigint," +
                            "   province_name string, " +
                            "   province_area_code string, " +
                            "   province_iso_code string, " +
                            "   province_3166_2_code string, " +
                            "   split_total_amount decimal(20,2), " +
                            "   order_id bigint, " +
                            "   create_time string, " +
                            "   et as to_timestamp(create_time), " +
                            "   watermark for et as et - interval '3' second" +  // string -> timestamp(3)
                            ")with(" +
                            "  'connector' = 'kafka', " +
                            "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                            "  'properties.group.id' = 'ProvinceStatsApp', " +
                            "  'topic' = '" + Constant.TOPIC_DWM_ORDER_WIDE + "', " +
                            "  'scan.startup.mode' = 'earliest-offset', " +
                            "  'format' = 'json'" +
                            ")");
        
        // 2. 开窗聚合
        Table result = tEnv.sqlQuery("select " +
                                         "    date_format(window_start, 'yyyy-MM-dd HH:mm:ss') stt, " +
                                         "    date_format(window_end, 'yyyy-MM-dd HH:mm:ss') edt, " +
                                         "    province_id, " +
                                         "    province_name, " +
                                         "    province_area_code area_code, " +
                                         "    province_iso_code iso_code, " +
                                         "    province_3166_2_code iso_3166_2," +
                                         "    sum(split_total_amount) order_amount, " +
                                         "    count(distinct(order_id)) order_count," +
                                         "    unix_timestamp()*1000 as ts " +
                                         " from table( tumble( table ow, descriptor(et), interval '5' second ) ) " +
                                         " group by province_id, province_name, province_area_code, province_iso_code, province_3166_2_code, " +
                                         " window_start, window_end");
//        result.execute().print();
        
        
        // 3. 写入到CLickhouse中
        tEnv
            .toRetractStream(result, ProvinceStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink(Constant.CLICKHOUSE_DB, "province_stats_2022", ProvinceStats.class));
        
        
    }
}
